/*
 * Licensed to the Apache Software Foundation (ASF) under one or more
 * contributor license agreements.  See the NOTICE file distributed with
 * this work for additional information regarding copyright ownership.
 * The ASF licenses this file to You under the Apache License, Version 2.0
 * (the "License"); you may not use this file except in compliance with
 * the License.  You may obtain a copy of the License at
 *
 *     http://www.apache.org/licenses/LICENSE-2.0
 *
 * Unless required by applicable law or agreed to in writing, software
 * distributed under the License is distributed on an "AS IS" BASIS,
 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 * See the License for the specific language governing permissions and
 * limitations under the License.
 */
package org.apache.rocketmq.store.schedule;

import java.util.HashMap;
import java.util.Iterator;
import java.util.Map;
import java.util.Map.Entry;
import java.util.Timer;
import java.util.TimerTask;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.ConcurrentMap;
import org.apache.rocketmq.common.ConfigManager;
import org.apache.rocketmq.common.TopicFilterType;
import org.apache.rocketmq.common.constant.LoggerName;
import org.apache.rocketmq.common.message.MessageAccessor;
import org.apache.rocketmq.common.message.MessageConst;
import org.apache.rocketmq.common.message.MessageDecoder;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.running.RunningStats;
import org.apache.rocketmq.store.ConsumeQueue;
import org.apache.rocketmq.store.ConsumeQueueExt;
import org.apache.rocketmq.store.DefaultMessageStore;
import org.apache.rocketmq.store.MessageExtBrokerInner;
import org.apache.rocketmq.store.PutMessageResult;
import org.apache.rocketmq.store.PutMessageStatus;
import org.apache.rocketmq.store.SelectMappedBufferResult;
import org.apache.rocketmq.store.config.StorePathConfigHelper;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

// 它是用于Broker服务器下去检查Schedule延迟消息是否已到时, 就绪时写入CommitLog对象中
public class ScheduleMessageService extends ConfigManager {
    private static final Logger log = LoggerFactory.getLogger(LoggerName.STORE_LOGGER_NAME);

    // 记录着延迟Message对象存储topic名称, default = SCHEDULE_TOPIC_XXXX;
    public static final String SCHEDULE_TOPIC = "SCHEDULE_TOPIC_XXXX";
    // 记录定时任务首次延迟时间戳, default = 1s
    private static final long FIRST_DELAY_TIME = 1000L;
    // 记录ConsumerQueue对象已消费完时, 定时任务延迟时间戳, default = 100ms
    private static final long DELAY_FOR_A_WHILE = 100L;
    // 记录定时任务异常时, 向定时任务线程池中的延迟执行时间戳
    private static final long DELAY_FOR_A_PERIOD = 10000L;

    // 记录每个level延迟级别下所对应的延迟时间戳
    private final ConcurrentMap<Integer /* level */, Long/* delay timeMillis */> delayLevelTable =
        new ConcurrentHashMap<Integer, Long>(32);

    // 记录每个level延迟级别下的consumerQueueOffset消费偏移量
    private final ConcurrentMap<Integer /* level */, Long/* offset */> offsetTable =
        new ConcurrentHashMap<Integer, Long>(32);

    // 记录用于定时任务的调度对象
    private final Timer timer = new Timer("ScheduleMessageTimerThread", true);

    private final DefaultMessageStore defaultMessageStore;

    // 记录Broker服务器下的最大Message延迟级别
    private int maxDelayLevel;

    public ScheduleMessageService(final DefaultMessageStore defaultMessageStore) {
        this.defaultMessageStore = defaultMessageStore;
    }

    // 它是用于将queueId转换成DelayLevel
    public static int queueId2DelayLevel(final int queueId) {
        return queueId + 1;
    }

    // 它是用于将DelayLevel转换成QueueId
    public static int delayLevel2QueueId(final int delayLevel) {
        return delayLevel - 1;
    }

    // 它是用于返回<level-offset>下的运行状态信息<delayOffset-MaxOffset>
    public void buildRunningStats(HashMap<String, String> stats) {
        Iterator<Entry<Integer /* level */, Long/* offset */>> it = this.offsetTable.entrySet().iterator();
        while (it.hasNext()) {
            Entry<Integer, Long> next = it.next();
            // 计算QueueId属性值
            int queueId = delayLevel2QueueId(next.getKey());
            long delayOffset = next.getValue();
            // 计算SCHEDULE_TOPIC_XXXX - queueId下的consumeQueue对象中最大有效偏移量
            long maxOffset = this.defaultMessageStore.getMaxOffsetInQueue(SCHEDULE_TOPIC, queueId);
            // 格式化delayOffset-MaxOffset
            String value = String.format("%d,%d", delayOffset, maxOffset);
            // 格式化scheduleMessageOffset - level
            String key = String.format("%s_%d", RunningStats.scheduleMessageOffset.name(), next.getKey());
            stats.put(key, value);
        }
    }

    // 更新delayLevel下的consumerQueue消费偏移量
    private void updateOffset(int delayLevel, long offset) {
        this.offsetTable.put(delayLevel, offset);
    }

    // 它是用于计算在delayLevel级别下Message对象多久时间才能开始消费
    public long computeDeliverTimestamp(final int delayLevel, final long storeTimestamp) {
        Long time = this.delayLevelTable.get(delayLevel);
        if (time != null) {
            return time + storeTimestamp;
        }

        return storeTimestamp + 1000;
    }

    // 启动ScheduleMessageService服务对象
    public void start() {

    	// 向定时任务线程池中添加定时任务, 延迟时间: default = 1s
        for (Map.Entry<Integer, Long> entry : this.delayLevelTable.entrySet()) {
            Integer level = entry.getKey();
            Long timeDelay = entry.getValue();
            Long offset = this.offsetTable.get(level);
            if (null == offset) {
                offset = 0L;
            }

            if (timeDelay != null) {
                this.timer.schedule(new DeliverDelayedMessageTimerTask(level, offset), FIRST_DELAY_TIME);
            }
        }

        // 向定时线程池中写入频率为10s的周期任务
        this.timer.scheduleAtFixedRate(new TimerTask() {

            @Override
            public void run() {
                try {
                	// 将消费偏移量写入磁盘中
                    ScheduleMessageService.this.persist();
                } catch (Throwable e) {
                    log.error("scheduleAtFixedRate flush exception", e);
                }
            }
        }, 10000, this.defaultMessageStore.getMessageStoreConfig().getFlushDelayOffsetInterval());
    }

    // shutdown 定时任务线程池
    public void shutdown() {
        this.timer.cancel();
    }

    public int getMaxDelayLevel() {
        return maxDelayLevel;
    }

    public String encode() {
        return this.encode(false);
    }

    // 从delayOffset.json文件中读取数据写入Map对象中
    public boolean load() {
        boolean result = super.load();
        result = result && this.parseDelayLevel();
        return result;
    }

    @Override // 获取delayOffset.json存储路径
    public String configFilePath() {
        return StorePathConfigHelper.getDelayOffsetStorePath(this.defaultMessageStore.getMessageStoreConfig()
            .getStorePathRootDir());
    }

    @Override // 将json格式的字符串对象写入至Map<Integer /* level */, Long/* offset */>对象中
    public void decode(String jsonString) {
        if (jsonString != null) {
            DelayOffsetSerializeWrapper delayOffsetSerializeWrapper =
                DelayOffsetSerializeWrapper.fromJson(jsonString, DelayOffsetSerializeWrapper.class);
            if (delayOffsetSerializeWrapper != null) {
                this.offsetTable.putAll(delayOffsetSerializeWrapper.getOffsetTable());
            }
        }
    }

    // 将level下消费偏移量Map对象编码成JSON格式的字符串对象
    public String encode(final boolean prettyFormat) {
        DelayOffsetSerializeWrapper delayOffsetSerializeWrapper = new DelayOffsetSerializeWrapper();
        delayOffsetSerializeWrapper.setOffsetTable(this.offsetTable);
        return delayOffsetSerializeWrapper.toJson(prettyFormat);
    }

    // 它是用于去解析Broker服务器下所对应Message延迟级所对应的延迟毫秒数
    public boolean parseDelayLevel() {
        HashMap<String, Long> timeUnitTable = new HashMap<String, Long>();
        // 1秒
        timeUnitTable.put("s", 1000L);
        // 1分钟
        timeUnitTable.put("m", 1000L * 60);
        // 1小时
        timeUnitTable.put("h", 1000L * 60 * 60);
        // 1天
        timeUnitTable.put("d", 1000L * 60 * 60 * 24);
        
        // "1s 5s 10s 30s 1m 2m 3m 4m 5m 6m 7m 8m 9m 10m 20m 30m 1h 2h";
        String levelString = this.defaultMessageStore.getMessageStoreConfig().getMessageDelayLevel();
        try {
            String[] levelArray = levelString.split(" ");
            for (int i = 0; i < levelArray.length; i++) {
                String value = levelArray[i];
                // 计算单位
                String ch = value.substring(value.length() - 1);
                // 计算该单位所对应的秒数
                Long tu = timeUnitTable.get(ch);

                int level = i + 1;
                if (level > this.maxDelayLevel) {
                    this.maxDelayLevel = level;
                }
                
                // 获取延迟级别下的数值部分
                long num = Long.parseLong(value.substring(0, value.length() - 1));
                // 计算延迟级别所对应的毫秒数
                long delayTimeMillis = tu * num;
                this.delayLevelTable.put(level, delayTimeMillis);
            }
        } catch (Exception e) {
            log.error("parseDelayLevel exception", e);
            log.info("levelString String = {}", levelString);
            return false;
        }

        return true;
    }

    class DeliverDelayedMessageTimerTask extends TimerTask {
    	// 记录着延迟级别, 即QueueId
        private final int delayLevel;
        // 记录着SCHEDULE_TOPIC_XXXX - QueueId下的消费偏移量
        private final long offset;

        public DeliverDelayedMessageTimerTask(int delayLevel, long offset) {
            this.delayLevel = delayLevel;
            this.offset = offset;
        }

        @Override
        public void run() {
            try {
                this.executeOnTimeup();
            } catch (Exception e) {
                log.error("ScheduleMessageService, executeOnTimeup exception", e);
                ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(
                    this.delayLevel, this.offset), DELAY_FOR_A_PERIOD);
            }
        }

        /**
         * @return
         */
        // ------------now--------------(now + delayTimeMillis)-----------
        // deliver < 0  |   deliver > 0      |       deliver = 0
        // 延迟时间已到                       延迟时间未到                                             延迟时间未到
        private long correctDeliverTimestamp(final long now, final long deliverTimestamp) {

            long result = deliverTimestamp;

            long maxTimestamp = now + ScheduleMessageService.this.delayLevelTable.get(this.delayLevel);
            if (deliverTimestamp > maxTimestamp) {
                result = now;
            }

            return result;
        }

        public void executeOnTimeup() {
        	// 获取该SCHEDULE_TOPIC - delayLevel(QueueId)下的ConsumeQueue对象
            ConsumeQueue cq =
                ScheduleMessageService.this.defaultMessageStore.findConsumeQueue(SCHEDULE_TOPIC,
                    delayLevel2QueueId(delayLevel));

            long failScheduleOffset = offset;

            if (cq != null) {
                SelectMappedBufferResult bufferCQ = cq.getIndexBuffer(this.offset);
                if (bufferCQ != null) {
                    try {
                        long nextOffset = offset;
                        int i = 0;
                        ConsumeQueueExt.CqExtUnit cqExtUnit = new ConsumeQueueExt.CqExtUnit();
                        // 遍历所有的字节数组对象, 以20字节为单位
                        for (; i < bufferCQ.getSize(); i += ConsumeQueue.CQ_STORE_UNIT_SIZE) {
                            // 获取CommitLog偏移量
                        	long offsetPy = bufferCQ.getByteBuffer().getLong();
                            // 获取CommitLog偏移量处的MessageExt对象所占用字节数
                        	int sizePy = bufferCQ.getByteBuffer().getInt();
                        	// 获取MessageExt对象下的Tag哈希值|ConsumerQueueExt偏移量
                            long tagsCode = bufferCQ.getByteBuffer().getLong();

                            if (cq.isExtAddr(tagsCode)) {
                            	// 获取CqExtUnit对象, 并获取MessageExt对象下TAG哈希值
                                if (cq.getExt(tagsCode, cqExtUnit)) {
                                    tagsCode = cqExtUnit.getTagsCode();
                                } else {
                                    //can't find ext content.So re compute tags code.
                                    log.error("[BUG] can't find consume queue extend file content!addr={}, offsetPy={}, sizePy={}",
                                        tagsCode, offsetPy, sizePy);
                                    // 获取消息的存储时间戳
                                    long msgStoreTime = defaultMessageStore.getCommitLog().pickupStoreTimestamp(offsetPy, sizePy);
                                    tagsCode = computeDeliverTimestamp(delayLevel, msgStoreTime);
                                }
                            }
                            
                            // 获取当前时间戳
                            long now = System.currentTimeMillis();
                            // 计算需要继续延迟时间戳
                            // <= 0 : 说明MessageExt对象已就绪, 即获取OriginTopic发送至CommitLog等待客户端消费
                            //  > 0 : 说明MessageExt对象还未就绪, 继续添加至定时任务线程池中去
                            long deliverTimestamp = this.correctDeliverTimestamp(now, tagsCode);

                            // 计算下一个待消费MessageExt对象下的consumerQueueOffset偏移量
                            nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
                            long countdown = deliverTimestamp - now;

                            if (countdown <= 0) {
                            	// 获取MessageExt对象
                                MessageExt msgExt =
                                    ScheduleMessageService.this.defaultMessageStore.lookMessageByOffset(
                                        offsetPy, sizePy);

                                if (msgExt != null) {
                                    try {
                                    	// 将Schedule-topic下的MessageExt对象转变topic下的MessageExt对象
                                        MessageExtBrokerInner msgInner = this.messageTimeup(msgExt);
                                        // 将MessageExt对象写入CommitLog对象中
                                        PutMessageResult putMessageResult =
                                            ScheduleMessageService.this.defaultMessageStore
                                                .putMessage(msgInner);

                                        // true : PUT_OK
                                        if (putMessageResult != null
                                            && putMessageResult.getPutMessageStatus() == PutMessageStatus.PUT_OK) {
                                            continue;
                                        } else {
                                            log.error(
                                                "ScheduleMessageService, a message time up, but reput it failed, topic: {} msgId {}",
                                                msgExt.getTopic(), msgExt.getMsgId());
                                            // 向定时任务线程池中写入延迟为10000L的定时任务
                                            ScheduleMessageService.this.timer.schedule(
                                                new DeliverDelayedMessageTimerTask(this.delayLevel,
                                                    nextOffset), DELAY_FOR_A_PERIOD);
                                            // 更新schedule下的消费偏移量
                                            ScheduleMessageService.this.updateOffset(this.delayLevel,
                                                nextOffset);
                                            return;
                                        }
                                    } catch (Exception e) {
                                        log.error(
                                            "ScheduleMessageService, messageTimeup execute error, drop it. msgExt="
                                                + msgExt + ", nextOffset=" + nextOffset + ",offsetPy="
                                                + offsetPy + ",sizePy=" + sizePy, e);
                                    }
                                }
                            } else {
                            	// 向定时任务线程池中写入延迟countdown时间戳的定时任务对象
                                ScheduleMessageService.this.timer.schedule(
                                    new DeliverDelayedMessageTimerTask(this.delayLevel, nextOffset),
                                    countdown);
                                // 更新schedule下的消费偏移量
                                ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);
                                return;
                            }
                        } // end of for

                        // 向定时任务线程池中写入延迟100ms的定时任务对象
                        nextOffset = offset + (i / ConsumeQueue.CQ_STORE_UNIT_SIZE);
                        ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(
                            this.delayLevel, nextOffset), DELAY_FOR_A_WHILE);
                        // 更新ConsumeQueue对象下的消费偏移量对象
                        ScheduleMessageService.this.updateOffset(this.delayLevel, nextOffset);
                        return;
                    } finally {

                        bufferCQ.release();
                    }
                } // end of if (bufferCQ != null)
                else {
                	
                	// 更新ConsumerQueue中最小有效偏移量
                    long cqMinOffset = cq.getMinOffsetInQueue();
                    if (offset < cqMinOffset) {
                        failScheduleOffset = cqMinOffset;
                        log.error("schedule CQ offset invalid. offset=" + offset + ", cqMinOffset="
                            + cqMinOffset + ", queueId=" + cq.getQueueId());
                    }
                }
            } // end of if (cq != null)
            // 向定时任务线程池中写入延迟100ms的定时任务对象
            ScheduleMessageService.this.timer.schedule(new DeliverDelayedMessageTimerTask(this.delayLevel,
                failScheduleOffset), DELAY_FOR_A_WHILE);
        }

        // 它是用于将Schedule-topic下的MessageExt对象转变topic写入CommitLog对象中
        private MessageExtBrokerInner messageTimeup(MessageExt msgExt) {
            // 创建MessageExtBrokerInner对象
        	MessageExtBrokerInner msgInner = new MessageExtBrokerInner();
            msgInner.setBody(msgExt.getBody());
            msgInner.setFlag(msgExt.getFlag());
            MessageAccessor.setProperties(msgInner, msgExt.getProperties());

            // SINGLE_TAG | MULTI_TAG
            TopicFilterType topicFilterType = MessageExt.parseTopicFilterType(msgInner.getSysFlag());
            // 获取MessageExt对象下的TAG哈希值对象
            long tagsCodeValue =
                MessageExtBrokerInner.tagsString2tagsCode(topicFilterType, msgInner.getTags());
            msgInner.setTagsCode(tagsCodeValue);
            msgInner.setPropertiesString(MessageDecoder.messageProperties2String(msgExt.getProperties()));

            msgInner.setSysFlag(msgExt.getSysFlag());
            msgInner.setBornTimestamp(msgExt.getBornTimestamp());
            msgInner.setBornHost(msgExt.getBornHost());
            msgInner.setStoreHost(msgExt.getStoreHost());
            // 记录已消费次数
            msgInner.setReconsumeTimes(msgExt.getReconsumeTimes());

            // 设置wait = false
            msgInner.setWaitStoreMsgOK(false);
            // 将DELAY属性值清空
            MessageAccessor.clearProperty(msgInner, MessageConst.PROPERTY_DELAY_TIME_LEVEL);

            // 设置REAL_TOPIC属性值, 实际上的topic属性值
            msgInner.setTopic(msgInner.getProperty(MessageConst.PROPERTY_REAL_TOPIC));
            // 设置REAL_QID属性值, 实际上的QueueId
            String queueIdStr = msgInner.getProperty(MessageConst.PROPERTY_REAL_QUEUE_ID);
            int queueId = Integer.parseInt(queueIdStr);
            msgInner.setQueueId(queueId);

            return msgInner;
        }
    }
}
